
\chapter{基于C++11的线程池的设计与实现}{}

\section{概况}{}
多线程对于现代高性能计算是必不可少的。从C++11开始，标准库就内置了底层的多线程支持，可使用的结构包括std::thread等。但值得注意的是，std::thread()每次被调用时都会创建一个新线程，而创建线程有一个显著的性能开销。此外，可以创建比硬件核心数更多的线程以同时处理，但可能会导致严重的减速\textsuperscript{\cite{杨开杰2010线程池的多线程并发控制技术研究}}。在本章中，我们提出了一个C++11兼容的线程池实现，它通过一次性创建一个包含固定数量线程的线程池，然后重用相同的线程，以在池的整个生命周期内执行不同的任务的形式来避免这些问题。用户将要执行的任务提交到队列中。每当一个线程变得可用时，它就弹出一个任务并执行它。
除了std::thread之外，C++标准库还提供了更高层次的std::async，这可能在内部使用线程池，但这是不能保证的，事实上，目前只有std::async的MSVC实现使用了一个线程池，而GCC和Clang没有\textsuperscript{\cite{cpp17threadpool}}。使用本文的定制线程池类而不是std::async允许用户拥有更多的控制权、透明度和可移植性。在实际编程使用中，线程池不是由各个模块单独管理，所以无法保证调用时能够得到及时的处理，因此其内聚性不强，建议只将不紧急的异步任务抛给线程池执行。

\begin{figure}[h]
  \centering
  \includegraphics[scale=0.45]{figures/threadpool.pdf}
  \caption{线程池概况}
  \label{fig:threadpool}
\end{figure}


\section{接口能力}{}

	\subsection{执行简单的异步任务}{}

以下代码展示了如何异步执行一个简单的耗时任务，并在主线程中主动等待其返回结果。若不等待，主线程将先于工作线程终止，而工作线程最终由操作系统进行回收。
\begin{lstlisting}[language=c++, label={lst:easy-async}]
std::future<int> future;
future = ThreadPool::Instance().Execute([] (int i) -> int {
    // simulate time consuming task
    sleep(1);
    return i * i;
}, 3);
printf("async function return %d.\n", future.get());
\end{lstlisting}

这是最常用的异步调用接口。将某些不希望阻塞主线程的耗时任务放到工作线程来处理，任务被线程池调度执行后，调用者通过接口返回的std::future等待获取返回值（此处是可选择的，因为也可以是不带返回值或者不等待异步返回值）。

	\subsection{执行带序列标签的异步任务}{}
某些情况下，用户需要某一类的任务与主线程相隔离，但是又希望这些任务们本身之间是串行的。
以下代码展示了如何设置两类任务——任务$A$、任务$B$，以异步于主线程，同步于自身其他任务的形式运行\textsuperscript{\cite{葛东遥2020一种同步任务异步执行的方法和调度系统}}。
\begin{lstlisting}[language=c++, label={lst:serialtag-async}]
// Define the serial_tag of the tasks which you want to execute serially
int taskA_serial_tag = 1;
int taskB_serial_tag = 2;

for (int i = 0; i < 4; ++i) {
    // Passing the serial_tag as the first param
    ThreadPool::Instance().Execute(taskA_serial_tag, [=] {
        printf("task A%d running...\n", i);
    	sleep(1);
        printf("task A%d done!\n", i);
    });

    ThreadPool::Instance().Execute(taskB_serial_tag, [=] {
        printf("task B%d running...\n", i);
        sleep(1);
        printf("task B%d done!\n", i);
    });
}
printf("main thread done.\n");
\end{lstlisting}

正如你在上面所看到的，总共有8个任务(每个任务耗时1秒)被加入到了线程池中，在4个工作线程运行的前提下，它耗费了4秒才完成。而不是$8*1/4=2$秒。这是因为具有相同序列化标签的任务是串行执行的。

这样的需求在实际的工程实现中是很常见的。譬如考虑这些情况，这种接口能力就适用了：
\begin{enumerate}
    \item 一个异步执行的计时器任务每秒才将屏幕上的倒计时数字减一，它们必须串行执行，否则四线程并发将使得计时器的计时速度变为现实的四倍。
    \item 长时间不打开微信后会出现的大量的新消息需要插入数据库的情况，这些任务如果比较耗时，则为了不影响主线程的流畅性，开发者将它们作为异步任务，同时我们不希望不同的消息插入到数据库时并发，因为这可能导致潜在的不必要的线程安全问题。
\end{enumerate}

使用时仅仅需要调用者定义此类任务的序列化标签，线程池就会自动根据标签区分任务并调度合理它们。详细的调度算法参见后文。

	\subsection{执行定时任务}{}
以下代码展示了如何在将来的某个时间点执行一个给定的任务。
\begin{lstlisting}[language=c++, label={lst:timing-async}]
ThreadPool::Instance().ExecuteAfter(2000, [=] {
	printf("task A running...\n");
	sleep(1);
	printf("task A done!\n");
});
ThreadPool::Instance().ExecuteAfter(1000, [=] {
	printf("task B running...\n");
	sleep(1);
	printf("task B done!\n");
});
printf("main thread done.\n");
\end{lstlisting}

一个可能的输出及其时间节点是这样的：
\begin{lstlisting}[label={lst:output1}]
main thread done.
task B running...   # at 1st second
task B done!        # at 2nd second
task A running...
task A done!        # at 3th second
\end{lstlisting}

或者也可能是这样的：
\begin{lstlisting}[label={lst:output2}]
main thread done.
task B running...   # at 1st second
task A running...   # at 2nd second
task B done!
task A done!        # at 3th second
\end{lstlisting}

用户在调用时指定一个时间间隔和一个异步任务，在调用的那一刻之后直到指定时间间隔到达之前，各个线程照常执行其他任务的指令序列，线程池将等到指定的时间间隔经过之后，调度执行给定的任务。使用场景如：在手机客户端发起一个新的连接请求之后，注册一个延时一秒钟的任务到线程池来检测此网络请求是否完成（或者出错）并清理这个连接，若执行时判断出此网络请求仍在接收数据中，回到第一步再抛同样一个延时任务，直到请求完成或者超时。此种情景下不使用周期型任务是因为这种任务有一个明确的截止条件——网络请求完成，超时或出错。

	\subsection{执行周期型任务}{}
以下代码将使得程序以1秒的时间间隔周期性地打印指定字符串：
\begin{lstlisting}[language=c++, label={lst:timing-async}]
ThreadPool::Instance().ExecutePeriodic(1000, [=] {
    printf("task running...\n");
});
\end{lstlisting}

定期地在线程池中执行异步的任务，使用场景诸如：服务端的网络连接池中需要定时清理某些连接，这些连接通常因为网络异常、不是来自合法的客户端程序而鉴权失败。此类任务没有明确的截止条件——服务端处于长期执行这类任务，因此此类任务的生命周期是跟随整个服务端进程的。当进程退出时，线程池会自行销毁这些任务。


\section{实现原理}{}
	\subsection{异步任务的表示形式}{}
一个异步任务的消息体主要应该包含三个字段：
    \begin{enumerate}
    	\item 任务类型——这是何种任务。任务类型包含3种类型：期望立即被执行的任务、期望指定时间后被执行的任务、期望被周期性执行的任务。此处可对照上文接口能力一节理解。
    	\item 序列化标签：调用者期待有着同一个序列化标签的多个任务之间是串行执行的。不同序列化标签的任务，由于它们不互相影响，故它们之间属于线程安全的，因此他们可以并发地执行。若调用者不关心此，或者本意就是希望某一类任务之间并发执行，将使用-1来填充此字段。
    	\item 异步任务的创建时间戳：此字段主要用于执行定时任务和周期任务时使用，将在接口被调用时获取对应时刻的时间戳，打在当前的任务消息体上。
    	\item 时间间隔（周期）：此字段易于理解，表示调用者期望任务执行的等待时间或多次执行之间的时间间隔。
    \end{enumerate}

	\subsection{任务的存储方式}{}
任务集合使用链表的形式进行存储。线程池中的任务作为一种经常参与调度的结构，因此经常变化。并且此变化的特征是可能从中间部分截取出某一个任务。因此，若使用动态数组，频繁的删除中间的元素将会复制移动后面的所有任务，开销过大。且由于任务集不需要进行按下标快速查询（具体原因请参考后文调度算法），所以也不需要将各个任务存在一个连续的内存空间内。综上所述，链表是目前最适合此情景且没有额外开销的数据结构。
	\subsection{调度策略}{}
同《操作系统原理》中进程调度章节的一个典型问题不同\textsuperscript{\cite{王刚2007Linux}}，线程池在进行任务调度时，不仅要考虑以何种方式选取任务，还要考虑的问题包括多个工作线程的合作问题，即同步与互斥——单个工作线程何时应该等待任务到来、何时应该以给定的方式去任务集合取任务。

互斥量保证任务链表的线程安全性\textsuperscript{\cite{孙明鹏2015浅谈多线程安全}}。首先解决的是多个线程访问共享变量即任务链表时的互斥操作，为了保证线程安全性，线程池应当避免来自由于某些读写操作的非原子性以及用于编译器优化的指令重排的产生的潜在问题，某一时刻只允许一个线程读/写任务链表。实现上使用C++11标准库的std::mutex配合std::lock\_guard保证。

互斥锁配合条件变量实现线程间通信。将分别从以下三种从简单到复杂的情况分析如何解决工作线程之间的通信问题。

首先考虑最简单的情况，所有任务都是期望立即被执行的任务，且不带序列化标签。闲置的情况下，各个工作线程不占用对任务队列的互斥量，等待有新的任务加入进来并且取出执行。注意此处的等待新任务到来不应是忙等（busy-wait），后者会导致占据了cpu时间片但没有完成任何有意义的工作、同时可能导致其他的工作线程拿不到互斥锁而出现饥饿现象。因不能采取忙等的机制，则单个工作线程必定不可能主动验证（或查询）是否有新的任务到来。 因此，必须在调用者向线程池注册新的任务时，从调用者的线程中主动地使用条件变量通知工作线程去检索任务链表。检索时拿到对任务链表的互斥锁后取出任务。同时，为了提高线程池的并发度，应尽可能地减小锁的粒度，比如在此种情况下，执行异步任务的过程中就应该释放锁，从而让其他的工作线程能尽快地去任务链表获取新任务。

出现带上序列化标签任务。

随着任务的种类逐渐复杂化，各个工作线程等待的条件就不再是之前的简单情况——即只要出现新任务就打破等待。由于带着同一个序列化标签的任务不能并发执行，在检测时，必须再判断其他工作线程正执行的任务中是否含有带同一序列化标签的任务，若有则不能立即执行该任务。同时还需要考虑执行完某一个带序列化标签的任务之后是否通知其他工作线程带此序列标签的任务可继续被执行。

出现定时/周期型任务。

这是最复杂的情况，工作线程的等待条件愈发难以被考虑周全。考虑有两种情况需要等待：

\begin{enumerate}
    \item 当前线程没有拿到任何任务，等待着新任务的到来。
    \item 当前线程已经拿到了一个定时任务，但是此任务还没有到达执行的时间点。
\end{enumerate}

第一种情况已经在上文中进行了详细的讨论，本文主要关注第二种情况。当前线程已经拿到了一个定时任务后，工作线程可以使用C++标准库的condition\_variable来实现定期等待。此处的难点在于，若在等待定时任务到期的过程中，调用者们又抛来了一个新的期望立即执行的任务（或者到期时间比当前持有的任务更短的任务）该怎样处理。

显然，线程池必不能等待现在的任务到期后才去执行刚刚的任务，因为这直接违背了线程池对外的提供的接口能力。此时，本文为任务引入了优先级的概念。规定线程池中到期时间较短的任务拥有相对较高的优先级，因此用户期望被立即执行的任务拥有最高的优先级。

基于此，本文应当将上述的"等待条件"从简单的等待新任务的到来，更改为等待更高优先级的任务到来。新任务插入时，通知工作线程检测现有任务链表，找出是否拥有比该工作线程当前持有的任务更高优先级的任务。若有，将当前持有的任务放回链表中，并将高优先级的任务取出，并等待其到期并执行（若立即执行类任务则不等待），若等待过程又出现更高优先级的任务，需立即重复执行上述流程。

